package com.itheima.kafka.example1;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * 示例1：简单的消息消费者
 * 
 * 学习目标：
 * 1. 如何使用 @KafkaListener 监听消息
 * 2. 理解消费者组的概念
 * 3. 如何处理接收到的消息
 */
@Slf4j
@Component
public class SimpleMessageListener {

    /**
     * 监听 simple-messages Topic 的消息
     * 
     * @KafkaListener 注解说明：
     * - topics: 要监听的Topic名称
     * - groupId: 消费者组ID（相同组ID的消费者会分担消息）
     */
    @KafkaListener(topics = "simple-messages", groupId = "simple-group")
    public void listen(ConsumerRecord<String, String> record) {
        log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
        log.info("📥 [消费者] 接收到新消息！");
        log.info("📥 [消费者] Topic: {}", record.topic());
        log.info("📥 [消费者] 分区: {}", record.partition());
        log.info("📥 [消费者] 偏移量: {}", record.offset());
        log.info("📥 [消费者] Key: {}", record.key());
        log.info("📥 [消费者] 消息内容: {}", record.value());
        log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
        
        // 这里可以添加你的业务逻辑
        // 比如：保存到数据库、调用其他服务等
    }
    
    /**
     * 另一个消费者（同一个消费者组）
     * 演示：同组的多个消费者会分担消息
     */
    @KafkaListener(topics = "simple-messages", groupId = "simple-group")
    public void listen2(ConsumerRecord<String, String> record) {
        log.info("👤 [消费者2] 我也接收到消息了！分区: {}, 消息: {}", record.partition(), record.value());
    }
}

